package com.springboot.zookeeper;

import com.springboot.config.ConsumerConfig;
import org.I0Itec.zkclient.ZkClient;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * zk配置文件
 *
 * @author supu
 * @create 2018-12-27 19:24
 **/
@Configuration
public class ZookeeperConfig {

    @Autowired
    private ConsumerConfig consumerConfig;

    @Bean
    public ZkClient zkClient(){
        String address = consumerConfig.getString(ConsumerConfig.ZOOKEEPER_ADDRESS);
        int sessionTimeout = Integer.valueOf(consumerConfig.getString(ConsumerConfig.ZOOKEEPER_SESSION_TIMEOUT));
        int connectionTimeout = Integer.valueOf(consumerConfig.getString(ConsumerConfig.ZOOKEEPER_CONNECTION_TIMEOUT));
        return new ZkClient(address,sessionTimeout, connectionTimeout, new CustomZookeeperSerializer());
    }

    @Bean
    public CuratorFramework curatorFramework(){
        String address = consumerConfig.getString(ConsumerConfig.ZOOKEEPER_ADDRESS);
        int sleepTime = Integer.valueOf(consumerConfig.getString(ConsumerConfig.ZOOKEEPER_CLIENT_SLEEP));
        int maxRetryCount = Integer.valueOf(consumerConfig.getString(ConsumerConfig.ZOOKEEPER_MAX_RETRY_COUNT));
        int sessionTimeout = Integer.valueOf(consumerConfig.getString(ConsumerConfig.ZOOKEEPER_SESSION_TIMEOUT));
        int connectionTimeout = Integer.valueOf(consumerConfig.getString(ConsumerConfig.ZOOKEEPER_CONNECTION_TIMEOUT));

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(sleepTime, maxRetryCount);
        CuratorFramework client =
                CuratorFrameworkFactory.builder()
                        .connectString(address)
                        .sessionTimeoutMs(sessionTimeout)
                        .connectionTimeoutMs(connectionTimeout)
                        .retryPolicy(retryPolicy)
                        .build();
        client.start();
        return client;
    }
}
